package main

import (
	"fmt"
	"log"
	"os"

	"github.com/colinmarc/hdfs"
)

// HDFS相关路径常量
const (
	hdfsBaseDir     = "/1"
	hdfsTargetFile  = "/1/AppleStore.csv"
	localSourceFile = "AppleStore.csv"
)

// createHDFSClient 创建一个新的HDFS客户端
// address: HDFS NameNode地址，格式如 "localhost:9000"
func createHDFSClient(address string) (*hdfs.Client, error) {
	client, err := hdfs.New(address)
	if err != nil {
		return nil, fmt.Errorf("failed to create HDFS client: %w", err)
	}

	return client, nil
}

// createHDFSClientWithUser 创建一个指定用户的HDFS客户端
// address: HDFS NameNode地址，格式如 "localhost:9000"
// user: 用户名
func createHDFSClientWithUser(address, user string) (*hdfs.Client, error) {
	client, err := hdfs.NewForUser(address, user)
	if err != nil {
		return nil, fmt.Errorf("failed to create HDFS client for user %s: %w", user, err)
	}

	return client, nil
}

// uploadCSVToHDFS 将本地CSV文件上传到HDFS
// client: HDFS客户端
// localFilePath: 本地CSV文件路径
// hdfsFilePath: HDFS目标文件路径
func uploadCSVToHDFS(client *hdfs.Client, localFilePath, hdfsFilePath string) error {
	// 使用CopyToRemote方法将本地文件上传到HDFS
	err := client.CopyToRemote(localFilePath, hdfsFilePath)
	if err != nil {
		return fmt.Errorf("failed to upload file %s to HDFS path %s: %w", localFilePath, hdfsFilePath, err)
	}

	return nil
}

// isHDFSPathExists 检查HDFS中路径是否存在
// client: HDFS客户端
// hdfsPath: HDFS路径
func isHDFSPathExists(client *hdfs.Client, hdfsPath string) (bool, error) {
	_, err := client.Stat(hdfsPath)
	if err != nil {
		if os.IsNotExist(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// createHDFSDir 创建HDFS目录
// client: HDFS客户端
// dirPath: 要创建的目录路径
func createHDFSDir(client *hdfs.Client, dirPath string) error {
	// 使用MkdirAll创建目录及其所有必要的父目录
	err := client.MkdirAll(dirPath, 0755)
	if err != nil {
		return fmt.Errorf("failed to create HDFS directory %s: %w", dirPath, err)
	}
	return nil
}

// listHDFSFiles 列出HDFS中指定目录下的文件
// client: HDFS客户端
// dirPath: HDFS目录路径
func listHDFSFiles(client *hdfs.Client, dirPath string) error {
	files, err := client.ReadDir(dirPath)
	if err != nil {
		return fmt.Errorf("failed to read directory %s: %w", dirPath, err)
	}

	fmt.Printf("Contents of directory %s:\n", dirPath)
	for _, file := range files {
		fileType := "file"
		if file.IsDir() {
			fileType = "dir"
		}
		fmt.Printf("%-8s %s\n", fileType, file.Name())
	}

	return nil
}

func main() {
	// 示例：创建一个HDFS客户端
	client, err := createHDFSClient("172.18.0.1:8020")
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	// 确保目标目录存在
	err = createHDFSDir(client, hdfsBaseDir)
	if err != nil {
		log.Fatal(err)
	}

	// 检查HDFS路径是否存在
	exists, err := isHDFSPathExists(client, hdfsTargetFile)
	if err != nil {
		log.Println("Error checking HDFS path:")
		log.Fatal(err)
	}

	// 如果文件存在，则先删除
	if exists {
		fmt.Println("File already exists in HDFS, it will be overwritten")
	}

	// 示例：上传CSV文件到HDFS
	err = uploadCSVToHDFS(client, localSourceFile, hdfsTargetFile)
	if err != nil {
		log.Println("Error uploading CSV file:")
		log.Fatal(err)
	}

	fmt.Println("Successfully uploaded CSV file to HDFS")

	// 展示HDFS目录中的文件
	err = listHDFSFiles(client, hdfsBaseDir)
	if err != nil {
		log.Fatal(err)
	}
}
